package com.atguigu.sparkstreaming.examples

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, KafkaUtils, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Created by Smexy on 2022/7/15
 *
 *    at least once + 幂等输出 = exactly once
 *
 *
 *    数据库按照 特性可以分为 两种:
 *
 *          Hive:  基于OLAP特性的数据仓库！
 *
 *          数据仓库：   历史版本留存
 *          数据库  ：   只保留最新版本
 *
 *          OLAP数据库:  online  analytic  process   在线分析(查询)处理。
 *                          hbase(filter,aggre)，es(DSL),clickhose,TiDB。
 *                          Presto，Impala，
 *
 *                          Nosql！  分布式！ 最难实现的就是事务！ 部分事务！
 *                          hbase:  如果要写入的数据都在一行，保证事务!
 *
 *
 *          OLTP数据库:  online transcation process  在线事务处理
 *                                RDMS:  mysql,oracle,DB2,sql server, Postgre Sql
 *
 *    数据库按照模式(是不是关系数据库)分为两类: NoSQL 和  RDMS
 *
 *
 */
object ExactlyOnceTemplateDemo {

  def main(args: Array[String]): Unit = {

    val streamingContext = new StreamingContext("local[*]", "TransformDemo", Seconds(10))

    //所有的消费者参数都可以在 ConsumerConfig
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "2203092",
      "auto.offset.reset" -> "latest",
      // 第一步
      "enable.auto.commit" -> "false"
    )

    val topics = Array("topicA")

    val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    //在Driver端声明偏移量
    var ranges: Array[OffsetRange] = null
    //第二步: 获取偏移量
    val ds1: DStream[ConsumerRecord[String, String]] = ds.transform(rdd => {

      ranges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

      //可以转换，在后续转换
      rdd
    })

    // 第三步：各种转换

    //使用转换后的ds
    /*ds2.foreachRDD(rdd => {

      //幂等输出到redis,mysql,es,hbase

      //手动提交offsets
      //选择一:  ds.asInstanceOf[CanCommitOffsets].commitAsync(ranges)  提交到kafka的 _comsumer_offsets
      //选择二:  将偏移量写入到任意数据库


    })*/

    // 启动APP
    streamingContext.start()

    // 阻塞进程，让进程一直运行
    streamingContext.awaitTermination()

  }

}
